package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 针对消息进行管理
 */
public class MessageFileManager {
    //定义一个内部类，来表示该队列的统计信息（静态内部类和外部类解耦）
    static public class Stat{
        //对于这样简单的类，直接使用成员，类似于C的结构体
        public int totalCount;//总消息数量
        public int validCount;//有效消息数量
    }

    public void init(){
        //暂时不需要初始化，先列在这
    }
    //预定消息文件所在的目录和文件名
    //这个方法，用来获取到指定队列对应的消息文件所在的路径
    //  data/testQueue1 data/testQueue2 data/testQueue3...
    private String  getQueueDir(String queueName){
        return "./data/"+queueName;
    }

    //这个方法用来获取该队列的消息数据文件路径
    //二进制文件使用txt作为后缀不太合适，但是先不改（.bin  .dat)
    //  data/testQueue1/queue_data.txt
    private String getQueueDataPath(String queueName){
        return getQueueDir(queueName)+"/queue_data.txt";
    }

    //这个方法用来获取该队列消息的统计路径
    //   data/testQueue1/queue_stat.txt
    private String getQueueStatPath(String queueName){
        return getQueueDir(queueName)+"/queue_stat.txt";
    }

    private Stat readStat(String queueName){
        //因为是文本文件，可以用Scanner读取
        Stat stat=new Stat();
        try(InputStream inputStream=new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner=new Scanner(inputStream);
            stat.totalCount=scanner.nextInt();
            stat.validCount=scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;

    }

    private void writeStat(String queueName,Stat stat){
        //使用PrintWriter写
        //OutputStream打开文件，默认会直接把原文件清空，相当于覆盖
        try(OutputStream outputStream=new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter=new PrintWriter(outputStream);
            printWriter.write(stat.totalCount+"\t"+stat.validCount);
            printWriter.flush();
        }  catch (IOException e) {
            e.printStackTrace();
        }
    }

    //创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        //1.先创建队列对应的消息目录
        File baseDir=new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            //不存在就创建这个目录
            boolean ok = baseDir.mkdirs();
            if(!ok){
                throw new IOException("创建目录失败！ baseDir="+baseDir.getAbsolutePath());
            }
        }
        //2.创建队列数据库文件
        File queueDataDir=new File(getQueueDataPath(queueName));
        if(!queueDataDir.exists()){
            boolean ok = queueDataDir.createNewFile();
            if(!ok){
                throw new IOException("创建数据文件失败！ queueDataDir="+queueDataDir.getAbsolutePath());
            }
        }
        //3.创建消息统计文件
        File queueStatDir=new File(getQueueStatPath(queueName));
        if(!queueStatDir.exists()){
            boolean ok = queueStatDir.createNewFile();
            if(!ok){
                throw new IOException("创建消息统计文件失败！ queueStatDir="+queueStatDir.getAbsolutePath());
            }
        }

        //4.给消息统计文件设定初始值，0\t0
        Stat stat=new Stat();
        stat.totalCount=0;
        stat.validCount=0;
        writeStat(queueName,stat);
    }

    //删除队列目录和文件
    //队列也是可以被删除的，当队列删除后，对应的文件啥的，也要随之删除
    public void destroyQueueFiles(String queueName) throws IOException {
        //先删除里面的文件，再删除对应的目录
        File queueDataPath=new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataPath.delete();
        File queueStatPath=new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatPath.delete();
        File baseDir=new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if(!ok1 || !ok2 || !ok3){
            //任意一个文件删除失败，整体删除失败
            throw new IOException("删除队列和目录文件失败！ baseDir="+baseDir.getAbsolutePath());
        }
    }

    //检查队列的目录和文件是否存在
    //后续有生产者给broker server 生产消息了，这个消息就可能需要记录到文件上（取决于消息是否持久化），所以需要判断文件是否存在，存在了才能记录
    public boolean checkFilesExits(String queueName){
        //判定数据文件和统计文件是否都存在
        File queueDataPath=new File(getQueueDataPath(queueName));
        if(!queueDataPath.exists()){
            return false;
        }
        File queueStatPath=new File(getQueueStatPath(queueName));
        if(!queueStatPath.exists()){
            return false;
        }
        return true;
    }

    //把一个新的消息，放到队列对应的文件中
    //queue表示要把这个消息写入队列，message则是要写的消息
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1.先检查当前要写入的队列对应的文件是否存在
        if (!checkFilesExits(queue.getName())) {
            throw new MqException("[MessageFileManager]队列对应的文件不存在！ queueName=" + queue.getName());
        }
        //2.把Message对象，进行序列化，转成二进制字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        //给队列加锁，为了防止多个线程同时进行写入，会产生冲突
        synchronized (queue) {
            //3.先获取到当前的队列数据文件的长度，用来计算出该 Message对象的 offsetBeg和 offsetEnd
            //把新的Message数据写入到队列数据文件的末尾
            //此时的 offsetBeg=当前文件长度+4   offsetEnd=当前文件长度+4+message自身的长度（约定好的）
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            //通过queueDataFile.length()就能获取到message的长度
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            //4.写入消息到数据文件，是需要追加的
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    //先写消息的长度，占4个字节，以int类型写入，所以使用 dataOutputStream.writeInt()
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息本体
                    dataOutputStream.write(messageBinary);
                }

            }
            //5.更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    //删除消息的方法
    //这里是逻辑上的删除，也就是把硬盘上存储的这个数据里面的那个isValid属性，设置成0
    //1.先把文件中的这一端数据，读出来，还原会Message对象
    //2.把 isValid改成0
    //3.把上述数据重新写会文件
    //此处这个参数中的message对象，必须包含有效的offsetBeg和offsetEnd
    public void deleteMessage(MSGQueue queue,Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                //1.先从文件读取对应的数据（ randomAccessFile.seek() 找到指定位置）
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                //2.把当前读取的二进制数据，转为 Message对象
                Message disMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                //3.把消息设为无效 isValid=0x0
                disMessage.setIsValid((byte) 0x0);
                //此处不需要给参数这个message的isValid设为0，因为这个参数代表的是内存中管理的message对象，而这个对象马上也要被从内存中销毁了
                //4.把消息转为二进制，重新写回文件
                //虽然上面已经seek过了，但是seek之后进行了读操作，导致光标往后移动，移动到下一个消息的位置了
                //想要回到之前的位置，就需要重新调整光标
                byte[] bufferDest = BinaryTool.toBytes(disMessage);
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }
            //不要忘了，更新统计文件
            //把一个消息设为无效了，此时有效消息就需要-1
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);

        }
    }

    //使用这个方法，从文件中，读取出所有的消息内容，加载到内存中（具体来说是放到一个链表里）
    //这个方法准备在程序启动的时候调用
    //这里使用一个LinkedList,主要目的是为了后序进行头删操作
    //这个方法的参数，只是一个queueName而不是MSGQueue对象，因为这个方法不需要加锁，只使用queueName就够了
    //由于这个方法是程序启动时调用的，测试服务器还不能处理请求，不涉及多线程操作文件
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        //使用这个变量记录当前文件光标
        long currentOffset=0;
        LinkedList<Message> messages=new LinkedList<>();
        try(InputStream inputStream=new FileInputStream(getQueueDataPath(queueName))){
            DataInputStream dataInputStream=new DataInputStream(inputStream);
                //一个文件中包含多条消息，需要循环读取
                while (true) {
                    //1.读取当前消息的长度,可能会读到文件末尾（EOF）
                    //  readInt方法，读到文件末尾，会出现EOFException异常，这一点和之前很多流对象不太一样
                    int messageSize = dataInputStream.readInt();
                    //按照这个长度，，读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if(messageSize!=actualSize){
                        //如果不匹配，说明文件有问题，格式错乱了
                        throw new MqException("[MessageFileManager]文件格式错误！ queueName="+queueName);
                    }
                    //3.把读到的二进制数据，反序列化回 Message
                    Message message = (Message)BinaryTool.fromBytes(buffer);
                    //4.判定一下这个消息对象是不是无效的
                    if(message.getIsValid()!=0x1){
                        //无效数据，直接跳过
                        //offset也要更新
                        currentOffset+=(4+messageSize);
                        continue;
                    }
                    //5.有效数据，则需要把这个Message对象加到链表中，还需要填写 offsetBeg和 offsetEnd
                    //  进行计算offset的时候，需要知道当前光标的位置，由于当前使用DataInputStream不方便计算当前光标的位置，所以需要手动计算
                    message.setOffsetBeg(currentOffset+4);
                    message.setOffsetEnd(currentOffset+4+messageSize);
                    currentOffset+=(4+messageSize);
                    messages.add(message);
            }
        }catch (EOFException e){
            //这个catch并非是正常异常，而是处理业务逻辑，表示到达文件末尾
            System.out.println("[MessageFileManager] 回复Message数据完成！");
        }
        return messages;
    }

    //检查当前是否要对该队列中的消息数据文件进行GC
    public boolean checkGC(String queueName){
        Stat stat=readStat(queueName);
        //规定总消息查过2000，并且有效消息占比少宇50%，就需要GC
        if(stat.totalCount>2000 && (double)stat.validCount/(double)stat.totalCount<0.5){
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName)+"/queue_data_new.txt";
    }

    //通过这个方法，真正执行消息数据文件的垃圾回收操作（复制算法）
    //创建一个新的文件，名字就是queue_data_new.txt
    //把之前消息数据文件中的有效消息都读取出来，写到新的文件中
    //删除旧的文件，再把新的文件改名会queue_data.txt
    //同时记得更新统计消息文件
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        //进行gc操作的时候，是针对消息数据文件的大洗牌，在这个过程中，其他线程不能对该队列进行操作
        synchronized (queue){
            //由于GC可能比较耗时间，所以此处统计一下消耗的时间
            long gcBeg=System.currentTimeMillis();
            //1.创建一个新的文件
            File queueDataNewFile=new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                //正常情况下这个不应该存在，如果存在说明上次gc了一半，程序意外崩溃了
                throw new MqException("[MessageFileManager] gc时发现该队列的queue_data_new.txt已经存在！queueName="+queue.getName());
            }
            boolean ok=queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建文件失败！queueDataNewFile="+queueDataNewFile.getAbsolutePath());
            }
            //2.从旧的文件读取出所有有效的消息
            LinkedList<Message> messages=loadAllMessageFromQueue(queue.getName());
            //3.把有效消息写到新文件
            try(OutputStream outputStream=new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream=new DataOutputStream(outputStream)){
                    for(Message message:messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //4.删除旧的数据文件，并把新的文件进行命名
            File queueDataOldFile=new File(getQueueDataPath(queue.getName()));
            ok=queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧的数据文件失败！ queueDataOldFile="+queueDataOldFile.getAbsolutePath());
            }
            //把 queue_data_new.txt 重命名回 queue_data.txt
            ok=queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败！ queueDataNewFile="+queueDataNewFile.getAbsolutePath()+",queueDataOldFile="+queueDataOldFile.getAbsolutePath());
            }
            //5.更新统计文件
            Stat stat=readStat(queue.getName());
            stat.totalCount=messages.size();
            stat.validCount=messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd=System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc执行完毕！ queueName="+queue.getName()+",time="+(gcEnd-gcBeg)+"ms");
        }
    }

}


















